package ex.tableapi;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import static org.apache.flink.table.api.Expressions.*;

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/tableapi/#groupby-window-aggregation
 *
 * 更换jar ：  https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc/3.1.1-1.17
 */
public class Ex06 extends ApiFrame {
    private String outName = "printOutTable";
    private String createPrintOutDDL = "CREATE TABLE printOutTable (" +
            "a STRING, " +
            "cnt  TIMESTAMP(3), " +
            "avgBillingAmount  BIGINT " +
            ") WITH (" +
            " 'connector' = 'print' " +
            ")";

    public static void main(String[] args) {
        Ex06 demo =new Ex06();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        demo.registerConsole(demo.createPrintOutDDL,demo.outName);
//        liji	500	tv	2024-01-22 19:03:55
//        liji	600	car	2024-01-23 19:04:01
//        liji	700	home	2024-01-24 19:04:19
//        liji	800	cloth	2024-01-24 19:04:24
//        zy	500	tv	2024-01-25 19:05:06
//        zy	600	car	2024-01-25 19:35:15
//        zy	700	home	2024-01-25 21:05:26
//        zy	800	cloth	2024-01-25 21:35:36
//TODO
        Table result = orders
                .filter(
                        and(
                                $("a").isNotNull(),
                                $("b").isNotNull(),
                                $("c").isNotNull()
                        ))
                .select($("a").lowerCase().as("a"), $("b"), $("rowtime"))
                .window(Tumble.over(lit(1).hour()).on($("rowtime")).as("hourlyWindow"))
                .groupBy($("hourlyWindow"), $("a"))
                .select($("a"), $("hourlyWindow").end().as("hour"), $("b").avg().as("avgBillingAmount"));
        result.printSchema();
//        1> +I[liji, 2024-01-23T04:00, 500]
//        1> +I[liji, 2024-01-24T04:00, 600]
//        1> +I[liji, 2024-01-25T04:00, 750]
//        1> +I[zy, 2024-01-26T04:00, 550]
//        1> +I[zy, 2024-01-26T06:00, 750]
        result.executeInsert(demo.outName);

    }
}
